package com.cx.base.day25_Rocketmq.producer.transaction;

import com.alibaba.druid.util.StringUtils;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.io.UnsupportedEncodingException;
import java.util.concurrent.TimeUnit;

public class SyncTransactionProducer {
    public static void main(String[] args) throws InterruptedException, RemotingException, MQClientException, MQBrokerException, UnsupportedEncodingException {
//        1.创建消息生产者，并且制定生产者组名 ,用TransactionMQProducer 替换掉 DefaultMQProducer
        TransactionMQProducer producer = new TransactionMQProducer("group1");
//        设置超时时间
        producer.setSendMsgTimeout(8000);
//        2.指定namesrv地址
        producer.setNamesrvAddr("192.168.48.128:9876;192.168.48.137:9876");
//        producer.setNamesrvAddr("47.96.165.93:9876;106.52.131.197:9876");
        //        设置事务监听器
        producer.setTransactionListener(new TransactionListener() {
            /**
             * 在此方法中执行本地事务
             * @param message
             * @param o
             * @return
             */
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {
                if (StringUtils.equals("tagA", message.getTags())) {
                    return LocalTransactionState.COMMIT_MESSAGE;
                } else if (StringUtils.equals("tagB", message.getTags())) {
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                } else if (StringUtils.equals("tagC", message.getTags())) {
                    return LocalTransactionState.UNKNOW;
                }
                return LocalTransactionState.UNKNOW;
            }

            /**
             * 此方法拥有MQ服务器回查时调用
             * @param messageExt
             * @return
             */
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                System.out.println("当前消息tag:" + messageExt.getTags());
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        });
//        3.启动producer
        try {
            producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }

        String[] tags = {"tagA", "tagB", "tagC"};
        for (int i = 0; i < 10; i++) {
//        4.创建消息对象，指定主题topic,tag和消息体
            Message msg = new Message("transactionTopic", tags[i % tags.length], ("hello world" + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
//        5.发送消息
            SendResult result = producer.sendMessageInTransaction(msg, null);
            String msgId = result.getMsgId();
            int queueId = result.getMessageQueue().getQueueId();
            System.out.println("send status:" + result + ",message ID:" + msgId + ",queue:" + queueId);
            TimeUnit.SECONDS.sleep(1);
            System.out.println(i);
        }
//        不要关闭producer，还要监听
//        producer.shutdown();

    }

}
